package com.huawei.mapreduce.mrsort;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueLineRecordReader;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.net.URI;

/**
 * 该类中，包括Mapper函数和Reducer函数共同实验了MapReduce的实现过程
 */
public class SecondSortMapReduce extends Configured implements Tool {

    public static void main(String[] args) {
        try {
            ToolRunner.run(new SecondSortMapReduce(), args);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class SecondSortMapper extends Mapper<Text, Text, CombinationKey, IntWritable> {
        /**
         * 这里要特殊说明一下，为什么要将这些变量写在map函数外边
         * 对于分布式的程序，我们一定要注意到内存的使用情况，对于MapReduce框架
         * 每一行的原始记录的处理都要调用一次map()函数，假设，这个map()函数要处理
         * 一亿条输入记录，如果将这些变量都定义在map函数里面则会导致这4个变量的对象句柄
         * 非常的多(极端情况下将产生4*1亿个句柄，当然java也是有自动的GC机制的，一定不会达到这么多)
         * 导致栈内存被浪费掉，我们将其写在map函数外面，顶多就只有4个对象句柄
         */
        private CombinationKey combinationKey = new CombinationKey();
        Text sortName = new Text();
        IntWritable score = new IntWritable();
        String[] splits = null;

        protected void map(Text key, Text value, Mapper<Text, Text, CombinationKey, IntWritable>.Context context) throws IOException, InterruptedException {
            System.out.println("---------------------进入map()函数---------------------");
            //过滤非法记录(这里用计数器比较好)
            if (key == null || value == null || key.toString().equals("")){
                return;
            }

            //构造相关属性
            sortName.set(key.toString());
            score.set(Integer.parseInt(value.toString()));
            //设置联合key
            combinationKey.setFirstKey(sortName);
            combinationKey.setSecondKey(score);
            //通过context把map处理后的结果输出
            context.write(combinationKey, score);
            System.out.println("---------------------结束map()函数---------------------");
        }
    }

    public static class SecondSortReducer extends Reducer<CombinationKey, IntWritable, Text, Text> {
        /**
         \* 这里要注意一下reduce的调用时机和次数：
         \* reduce每次处理一个分组的时候会调用一次reduce函数。
         \* 所谓的分组就是将相同的key对应的value放在一个集合中
         \* 例如：<sort1,1> <sort1,2>
         \* 分组后的结果就是
         \* <sort1,{1,2}>这个分组会调用一次reduce函数
         */
        protected void reduce(CombinationKey key, Iterable<IntWritable> values, Reducer<CombinationKey, IntWritable, Text, Text>.Context context)
                throws IOException, InterruptedException {
            //将联合Key的第一个元素作为新的key，遍历values将结果写出去
            for (IntWritable val : values){
                context.write(key.getFirstKey(), new Text(String.valueOf(val.get())));
            }
        }
    }

    public int run(String[] args) throws Exception {
        try {
            // 创建配置信息
            Configuration conf = new Configuration();
            conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPARATOR, "\t");

            // 创建任务
            Job job = Job.getInstance(conf, SecondSortMapReduce.class.getName());
            //设置成jar运行型
            job.setJarByClass(SecondSortMapReduce.class);

            //设置自定义Mapper类和设置map函数输出数据的key和value的类型
            job.setMapperClass(SecondSortMapper.class);
            job.setMapOutputKeyClass(CombinationKey.class);
            job.setMapOutputValueClass(IntWritable.class);

            //设置分区和reduce数量(reduce的数量，和分区的数量对应)
            job.setPartitionerClass(DefinedPartition.class);
            job.setNumReduceTasks(3);

            //设置自定义分组策略
            job.setGroupingComparatorClass(DefinedGroupSort.class);
            //设置自定义比较策略(因为CombineKey重写了compareTo方法，所以这个可以省略)
            //job.setSortComparatorClass(DefinedComparator.class);

            //指定Reducer类和输出key和value的类型
            job.setReducerClass(SecondSortReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);

            //设置输入目录和设置输入数据格式化的类
            FileInputFormat.setInputPaths(job, args[0]);
            job.setInputFormatClass(KeyValueTextInputFormat.class);

            // 创建文件系统
            FileSystem fs = FileSystem.get(new URI(args[1]), conf);
            // 如果输出目录存在，我们就删除
            if (fs.exists(new Path(args[1]))) {
                fs.delete(new Path(args[1]), true);
            }

            //指定输出的路径和设置输出的格式化类
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
            job.setOutputFormatClass(TextOutputFormat.class);

            // 提交作业 退出
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return 0;
    }
}